package org.apache.spark.doe

import org.apache.spark.{SecurityManager, SparkConf}
import org.apache.spark.rpc.{RpcAddress, RpcCallContext, RpcEndpointRef, RpcEnv, ThreadSafeRpcEndpoint}

class Worker( val rpcEnv: RpcEnv) extends ThreadSafeRpcEndpoint {

  var masterEndpoint: RpcEndpointRef = null

  override def onStart(): Unit = {
    //向Master发送建立连请求
    masterEndpoint = rpcEnv.setupEndpointRef(RpcAddress("localhost", 8888), "Master")
    //向Master发送注册Worker请求
    masterEndpoint.send(RegisterWorker("w0001", "localhost", 9999))
  }

  //用于接受异步请求的方法
  override def receive: PartialFunction[Any, Unit] = {
    case "Registered" =>
      println("Worker注册成功了")
  }

  //用于接受同步请求的方法
  override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
    case "response" =>
      println(888)
  }

}

object Worker {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf
    val securityMgr = new SecurityManager(conf)
    val rpcEnv = RpcEnv.create("SparkWorker", "localhost", 9999, conf, securityMgr)
    rpcEnv.setupEndpoint("Worker", new Worker(rpcEnv))
    rpcEnv.awaitTermination()
  }
}